"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 行存压缩
Case Name   : 事务内执行alter修改压缩参数，回滚/提交成功
Create At   : 2023/8/9
Owner       : opentestcase013
Description :
    1、建普通压缩表，并插入数据
    2、建分区压缩表，并插入数据
    3、建二级分区表，并插入数据
    4、事务中修改压缩参数,回滚
    5、事务中修改压缩参数,提交
    6、清理环境
Expect      :
    1、成功
    2、成功
    3、成功
    4、成功
    5、成功
    6、成功
History     :
"""

import os
import unittest
import re

from yat.test import Node

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class RowCompressTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.pri_user = Node('PrimaryDbUser')
        self.constant = Constant()
        self.tb_name = ['t_alter_rowcompress_0091_01',
                        't_alter_rowcompress_0091_02',
                        't_alter_rowcompress_0091_03']
        self.create_compress_params = ['compresstype=1', 'compresstype=2']
        self.alter_compress_params = ['compresstype=2, '
                                      'compress_byte_convert=true, '
                                      'compress_diff_convert=true, '
                                      'compress_level=16, '
                                      'compress_chunk_size=2048, '
                                      'compress_prealloc_chunks=3',
                                      'compresstype=0',
                                      'compresstype=1, '
                                      'compress_byte_convert=true, '
                                      'compress_diff_convert=true, '
                                      'compress_chunk_size=2048, '
                                      'compress_prealloc_chunks=2']

    def test_row_compress(self):
        text = '-----step1:建普通压缩表，并插入数据;expect:成功-----'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name[0]} cascade;
        create table {self.tb_name[0]}(columnone 
        integer, 
        columntwo char(50), columnthree varchar(50), columnfour integer, 
        columnfive char(50), columnsix varchar(50), columnseven char(50), 
        columneight char(50), columnnine varchar(50), columnten varchar(50), 
        columneleven char(50), columntwelve char(50), 
        columnthirteen varchar(50), columnfourteen char(50), 
        columnfifteem varchar(50)) with ({self.create_compress_params[0]});
        insert into {self.tb_name[0]} values(generate_series(0, 1000), 
        'test',substring(md5(random()::text), 1, 16), 
        2, substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16));
        '''
        sql_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn('INSERT 0 1001', sql_res, '执行失败：' + text)

        text = '-----step2:建分区压缩表，并插入数据;expect:成功-----'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name[1]} cascade;
                create table {self.tb_name[1]}
                (prod_id number(6),
                 cust_id number,
                 time_id date,
                 channel_id char(1),
                 promo_id number(6),
                 quantity_sold number(3),
                 amount_sold number(10,2)
                )
                with ({self.create_compress_params[1]})
                partition by range (time_id)
                interval('1 day')
                ( partition p1 values less than ('2023-02-01 00:00:00'),
                  partition p2 values less than ('2023-02-02 00:00:00')
                );
                insert into {self.tb_name[1]} values(generate_series(0, 1000), 
                lpad(floor(random() * 1000)::text, 2, '0'), 
                '2023-01-10 00:00:00', 'a', 1, 1, 1);
                insert into {self.tb_name[1]} values(generate_series(0, 1000), 
                lpad(floor(random() * 1000)::text, 2, '0'), 
                '2023-02-01 00:00:00', 'a', 2, 2, 2);
                '''
        sql_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn('INSERT 0 1001', sql_res, '执行失败：' + text)
        self.assertEqual(sql_res.count('INSERT 0 1001'), 2, '执行失败：' + text)

        text = '-----step3:建二级分区表，并插入数据;expect:成功-----'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name[2]} cascade;
                create table {self.tb_name[2]}
                (
                    month_code varchar2 ( 30 ) not null ,
                    dept_code  varchar2 ( 30 ) not null ,
                    user_no    varchar2 ( 30 ) not null ,
                    sales_amt  int
                )
                partition by list (month_code) subpartition by list (dept_code)
                (
                  partition p_201901 values ( '201902' )
                  (
                    subpartition p_201901_a values ( '1' ),
                    subpartition p_201901_b values ( '2' )
                  ),
                  partition p_201902 values ( '201903' )
                  (
                    subpartition p_201902_a values ( '1' ),
                    subpartition p_201902_b values ( '2' )
                  )
                );
                insert into {self.tb_name[2]} values('201902', '1', '1', 1);
                insert into {self.tb_name[2]} values('201902', '2', '1', 1);
                insert into {self.tb_name[2]} values('201902', '1', '1', 1);
                insert into {self.tb_name[2]} values('201903', '2', '1', 1);
                insert into {self.tb_name[2]} values('201903', '1', '1', 1);
                insert into {self.tb_name[2]} values('201903', '2', '1', 1);
                '''
        sql_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn('INSERT 0 1', sql_res, '执行失败：' + text)
        self.assertEqual(sql_res.count('INSERT 0 1'), 6, '执行失败：' + text)

        text = '-----step4:事务中修改压缩参数,回滚;expect:成功-----'
        self.log.info(text)
        sql = f'''begin;
        alter table t_alter_rowcompress_0091_01 set 
        ({self.alter_compress_params[0]});
        checkpoint;
        alter table t_alter_rowcompress_0091_02 set 
        ({self.alter_compress_params[1]});
        checkpoint;
        alter table t_alter_rowcompress_0091_03 set 
        ({self.alter_compress_params[2]});
        rollback;'''
        res = self.pri_sh.execut_db_sql(sql)
        self.log.info(res)
        for i in range(3):
            sql = f'\\d+ {self.tb_name[i]}'
            res = self.pri_sh.execut_db_sql(sql)
            self.log.info(res)
            if i < 2:
                self.assertIn(self.create_compress_params[i], res,
                              '执行失败：' + text)
            else:
                self.assertNotIn('compresstype', res, '执行失败：' + text)

        text = '-----step5:事务中修改压缩参数,提交;expect:成功-----'
        self.log.info(text)
        sql = f'''begin;
        alter table t_alter_rowcompress_0091_01 set 
        ({self.alter_compress_params[0]});
        checkpoint;
        alter table t_alter_rowcompress_0091_02 set 
        ({self.alter_compress_params[1]});
        checkpoint;
        alter table t_alter_rowcompress_0091_03 set 
        ({self.alter_compress_params[2]});
        commit;
        '''
        res = self.pri_sh.execut_db_sql(sql)
        self.log.info(res)
        for i in range(3):
            sql = f'\\d+ {self.tb_name[i]}'
            res = self.pri_sh.execut_db_sql(sql)
            self.log.info(res)
            self.assertIn(self.alter_compress_params[i], res)

    def tearDown(self):
        text = '-----step6:清理环境; expect:成功-----'
        self.log.info(text)
        drop_res = []
        for k, v in enumerate(self.tb_name):
            sql = f'drop table if exists {v};'
            drop_res.append(self.pri_sh.execut_db_sql(sql))
            self.log.info(drop_res)
            self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_res[k],
                          '执行失败：' + text)
            k += 1
        self.assertEqual(
            sum(1 for s in drop_res if re.search(r'DROP\s+TABLE', s)), 3,
            '执行失败：' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')